package demo.websocket.server.example3.rocketmq;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQReplyListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.stereotype.Component;

import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.Map;

/**
 * @author zhangjg
 * @desc
 * @date 2023/05/27 15:24
 **/
@Component
@RocketMQMessageListener(topic = "userTopic", consumerGroup = "p1", selectorExpression = "*")
public class SubUserTopic implements RocketMQListener<MessageExt> {

    private static final Logger logger = LoggerFactory.getLogger(SubUserTopic.class);

    @Autowired
    private SimpMessagingTemplate simpMessagingTemplate;

    @Override
    public void onMessage(MessageExt messageExt) {
        Map<String, String> properties = messageExt.getProperties();
        String sessionAttr = properties.get("simpSessionAttributes");
        String jsonStr = sessionAttr
                .replace("{","{\"")
                .replace("}","\"}")
                .replaceAll("=", "\":\"")
                .replaceAll(", ", "\",\"");
        JSONObject attr = JSON.parseObject(jsonStr);
        String fromUserId = attr.getString("userId");
        String toUserId = attr.getString("toUserId");
        String body = new String(messageExt.getBody(), StandardCharsets.UTF_8);
        simpMessagingTemplate.convertAndSendToUser(toUserId, "/msg", body);
        logger.info("订阅到--点对点--消息,fromUserId={},toUserId={},msg={}",fromUserId,toUserId, body);
    }
}
